use crate::logical_plan::expr::{LiteralValue, Operator};
use crate::physical_plan::{PhysicalPlanTrait, PhysicalPlanExpr};
use chrono::NaiveDateTime;
use crate::data_frame::{DataFrame, RealValue, Data, DataInstance};
use std::collections::VecDeque;
use std::ops::{Try, ControlFlow, FromResidual, Deref, DerefMut};
use async_recursion::async_recursion;

pub enum RunPlan{
    Filter{
        left: Box<RunPlanPhysicalPlanFuncArg>,
        op: Box<dyn PhysicalPlanTrait>,
        right: Box<RunPlanPhysicalPlanFuncArg>,
    },
    Func(Box<dyn PhysicalPlanTrait>,VecDeque<RunPlanPhysicalPlanFuncArg>),
    Not(Box<RunPlan>),
}

#[derive(Debug)]
pub enum RunPlanPhysicalPlanFuncArg{
    Normal(Box<PhysicalPlanExpr>),
    OtherRunPlan(Box<RunPlan>),
}

pub struct RunPlanInstance{
    pub name: String,
    pub run_plan: Box<RunPlan>,
}
impl RunPlanInstance{
    pub fn new(name: &str, run_plan: Box<RunPlan>)->Box<RunPlanInstance>{
        return Box::new(RunPlanInstance{
            name: name.to_string(),
            run_plan: run_plan,
        });
    }
}

impl RunPlanInstance{
    pub fn name(&self)->&str{
        return self.name.as_str();
    }
}

impl std::fmt::Debug for RunPlan {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self{
            RunPlan::Filter {left,op,right} => {
                f.debug_tuple("Filter")
                    .field(left)
                    .field(op)
                    .field(right)
                    .finish()
            },
            RunPlan::Func(physical_plan, args) => {
                f.debug_tuple("Func")
                    .field(&args)
                    .field(&physical_plan.name())
                    .finish()
            },
            RunPlan::Not(t) => {
                f.debug_tuple("Not").field(t).finish()
            },
        }
    }
}

impl RunPlanInstance{
    pub async fn evaluate(&mut self, data: &mut DataInstance)->HandResult{
        return self.run_plan.evaluate(data).await;
    }
}

impl RunPlan{
    #[async_recursion]
    pub async fn evaluate(&mut self, data: &mut DataInstance)->HandResult{
        match self{
            RunPlan::Filter {left,op,right} => {
                let left_expr = match left.deref_mut(){
                    RunPlanPhysicalPlanFuncArg::Normal(t) => {t.clone()},
                    RunPlanPhysicalPlanFuncArg::OtherRunPlan(t) => {
                        t.evaluate(data).await?.to_physical_expr()
                    },
                };
                let right_expr = match right.deref_mut(){
                    RunPlanPhysicalPlanFuncArg::Normal(t) => {t.clone()},
                    RunPlanPhysicalPlanFuncArg::OtherRunPlan(t) => {
                        t.evaluate(data).await?.to_physical_expr()
                    },
                };
                return op.evaluate(vec![left_expr,right_expr].as_slice(), data).await;
            },
            RunPlan::Func(func, args) => {
                let mut tmp = Vec::new();

                while let Some(arg) = args.pop_front(){
                    match arg{
                        RunPlanPhysicalPlanFuncArg::Normal(t) => {
                            tmp.push(t);
                        },
                        RunPlanPhysicalPlanFuncArg::OtherRunPlan(mut other) => {
                            let other_ret = other.evaluate(data).await?;
                            tmp.push(Box::new(PhysicalPlanExpr::RealValue(other_ret.value)));
                        },
                    }
                }
                return func.evaluate(tmp.as_slice(), data).await;
            },
            RunPlan::Not(t) => {
                let t_ret = t.evaluate(data).await?;
                if let RealValue::Boolean(value) = t_ret.value{
                    return HandResult::Ok(HandResultOk::new(RealValue::Boolean(!value)));
                }else{
                    return HandResult::Err(HandResultFail::new(format!("not plan with non Boolean result")));
                }
            },
        }
    }
}

pub struct HandResultOk{
    pub value: RealValue,
}
impl HandResultOk{
    pub fn new(value: RealValue)->HandResultOk{
        return HandResultOk{value};
    }
    pub fn into(self)->HandResult{
        return HandResult::Ok(self);
    }
    pub fn to_physical_expr(self)->Box<PhysicalPlanExpr>{
         return Box::new(PhysicalPlanExpr::RealValue(self.value));
    }
}
#[derive(Clone)]
pub struct HandResultFail{
    pub value: VecDeque<String>,
}
impl HandResultFail{
    pub fn new(desc: String)->HandResultFail{
        let mut value = VecDeque::new();
        value.push_back(desc);
        return HandResultFail{value};
    }
    pub fn append(&mut self, desc: String){
        self.value.push_back(desc);
    }
    pub fn into(self)->HandResult{
        HandResult::Err(self)
    }
}

pub enum HandResult{
    Ok(HandResultOk),
    Err(HandResultFail),
}
impl Try for HandResult{
    type Output = HandResultOk;
    type Residual = HandResultFail;
    fn branch(self)->ControlFlow<HandResultFail,HandResultOk>{
        match self{
            HandResult::Ok(t) => {ControlFlow::Continue(t)},
            HandResult::Err(t) => {ControlFlow::Break(t)},
        }
    }
    fn from_output(t: HandResultOk)->Self{
        HandResult::Ok(t)
    }
}

impl FromResidual for HandResult{
    fn from_residual(t: HandResultFail) -> Self{
        HandResult::Err(t)
    }
}

impl HandResult{
    pub fn new_fail(desc: String)->HandResult{
        return HandResultFail::new(desc).into();
    }
    pub fn new_ok(value: RealValue)->HandResult{
        return HandResultOk::new(value).into();
    }
}

